package com.alibaba.datax.core.transport.transformer;

import com.alibaba.datax.common.element.Column;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.element.StringColumn;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.core.util.geo.MapGeo;
import com.alibaba.datax.core.util.geo.amap.AMapGeo;
import com.alibaba.datax.core.util.geo.coogis.CooGISGeo;
import com.alibaba.datax.core.util.geo.response.GeoInfoResponse;
import com.alibaba.datax.transformer.Transformer;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;

/**
 * 地理编码转换
 *
 * @author wangyy
 */
public class GeocodingTransformer extends Transformer {

    private static final Logger LOG = LoggerFactory.getLogger(GeocodingTransformer.class);

    public GeocodingTransformer() {
        setTransformerName("dx_geocoding");
        LOG.info("Using geocoding preserve masker");
    }

    /**
     * paras：
     * 第一个参数:
     * geo 正向地理编码 、 rgeo 逆向地理编码
     * 第二个参数：
     * 正向地理编码时，配置地名地址字段编号，例子 ：2 ，代表编号为2的字段为地名地址列
     * 逆向地理编码时，经度纬度在同一字段的，配置字段编号 与 分隔符，例 :  '2@,'  代表编号为2的字段，经纬度使用‘,’分割。
     * 经纬度不在同一字段时,配置经纬度字段编号，经度在前纬度在后，例 ： '2@3' ，代表编号2的字段为经度，编号3的字段为纬度
     * <p>
     * 第三个参数：转换服务商，amap or coogis，非必选，默认coogis
     * 第四个参数：
     * 在使用coogis服务时，需要填写请求地址 ： http://xxx.xxx.xxx:xxxx
     * 在使用amap服务时，需要填写amap version and key ,最终使用@符号拼接, 例子 ： 'v3@9f2b4....42bfe'
     *
     * @param record 行记录，UDF进行record的处理后，更新相应的record
     * @param paras  transformer函数参数
     */
    @Override
    public Record evaluate(Record record, Object... paras) {
        // long startTs = System.currentTimeMillis();
        int columnIndex = (int) paras[0];
        //地理编码 or 逆向地理编码
        String isReverse = (String) paras[1];
        //字段编号
        String reverseColumn = (String) paras[2];
        String services = (String) paras[3];
        String url = (String) paras[4];
        MapGeo mapGeo;
        try {
            switch (services) {
                case "amap":
                    String[] split = url.split("@");
                    mapGeo = new AMapGeo(split[0], split[1]);
                    break;
                default: // "coogis"
                    mapGeo = new CooGISGeo(url);
                    break;
            }
        }catch (Exception e) {
            throw DataXException.asDataXException(TransformerErrorCode.TRANSFORMER_ILLEGAL_PARAMETER, "paras:" + Arrays.asList(paras).toString());
        }

        String result = "";
        if (isReverse.equals("geo")) {
            result = forwardGeocoding(reverseColumn, record, mapGeo);
        } else if (isReverse.equals("rgeo")) {
            result = reverseGeocoding(reverseColumn, record, mapGeo);
        }

        record.addColumn(new StringColumn(result));

        //record.setColumn(target, new StringColumn(result));
        // long diff = System.currentTimeMillis() - startTs;
        // LOG.info("==============================================  :" + diff);
        return record;
    }

    /**
     * 地理编码
     *
     * @param record
     * @param reverseColumn 结构化地址字段信息 必填
     *                      内部包含信息规则遵循：国家、省份、城市、区县、城镇、乡村、街道、门牌号码、屋邨、大厦，如：北京市朝阳区阜通东大街6号。
     * @return
     */
    public String forwardGeocoding(String reverseColumn, Record record, MapGeo mapGeo) {
        String address = "";
        if (reverseColumn != null && !reverseColumn.trim().isEmpty()) {
            Column column = record.getColumn(Integer.parseInt(reverseColumn));
            if (column != null) {
                address = column.asString();
            }
        }
        if (StringUtils.isBlank(address)) {
            return "";
        }
        GeoInfoResponse regeo = mapGeo.geo(address, "");
        if (regeo != null)
            return regeo.getLocation();
        return "";
    }

    /**
     * 逆地理编码
     *
     * @param reverseColumn
     * @param record
     * @param mapGeo
     * @return x, y
     */
    public String reverseGeocoding(String reverseColumn, Record record, MapGeo mapGeo) {
        String lon, lat;
        String[] split = new String[0];
        if (reverseColumn != null && !reverseColumn.trim().isEmpty() && reverseColumn.contains("@")) {
            split = reverseColumn.split("@");
            if (split.length < 2) {
                return "";
            }
        }
        String last = split[1];

        if (last.matches("-?\\d+(\\.\\d+)?")) {
            //经纬度在两个字段内
            lon = record.getColumn(Integer.parseInt(split[0])).asString();
            //如果字段为空，跳过subStr处理
            if (lon == null) {
                return "";
            }
            lat = record.getColumn(Integer.parseInt(split[1])).asString();
        } else {
            //经纬度在一个字段内，使用特定分隔符分割
            String asString = record.getColumn(Integer.parseInt(split[0])).asString();
            String[] split1 = asString.split(split[1]);
            lon = split1[0];
            lat = split1[1];
        }

        if (lon == null || lat == null) return "";

        String location = String.join(",", lon, lat);

        GeoInfoResponse regeo = mapGeo.regeo(location);

        if (regeo != null) {
            return regeo.getFormatted_address();
        }
        return "";
    }
}